04 Tools组件

在一个AI Agent智能体中，通过LLM来进行思考和推理，通过工具调用来真正的完成工作，本文将会详细如何使用LangChain提供的工具组件。

一、创建工具

3.1 @tool装饰器

创建工具最简单的方法就是通过@tool装饰器，示例如下，只需要定义一个普通的Python函数，并在方法上添加@tool，并且在注释中需要描述工具和参数的作用，以便方便LLM模型选择工具和生成工具参数。

@tool
def get_ip_location(ip: str) -> str:
    """通过ip地址获取位置信息

    Args:
        ip: 要查询定位的ip地址
    """
    return f"ip地址：{ip}，定位地址：浙江杭州"

还可以通过@tool装饰器去指定工具名称和工具描述。

@tool("search_ip_location", description="通过用户提供的ip地址，获取对应地理位置")
def get_ip_location(ip: str) -> str:
    """通过ip地址获取位置信息

    Args:
        ip: 要查询定位的ip地址
    """
    return f"ip地址：{ip}，定位地址：浙江杭州"

3.2 Schema定义

除了直接使用简单类型如str，int来作为工具参数，还可以使用Pydantic模型或者JSON来定义复杂的工具输入，推荐使用Pydantic模型，因为Pydantic模型可以帮我们完成一些数据校验工作。

from langchain.tools import tool
from pydantic import BaseModel, Field


class SearchUserInput(BaseModel):
    nickname: str = Field(description="用户名")
    sex: str = Field(description="用户性别")


@tool(args_schema=SearchUserInput)
def get_user_info(nickname: str, sex: str) -> str:
    return f"用户名：{nickname}, 性别：{sex}, 年龄：18, 简介：这个人很懒什么都没有留下"

二、ToolRunTime访问上下文

在LangChain v1.0版本中， LangChain支持通过ToolRuntime，在工具内部获取和更新上下文信息，上下文包括State、短期记忆、长期记忆、执行信息等数据，可以把ToolRunTime理解为整个Agent运行时上下文数据。

这里要注意的是通过ToolRuntime获取上下文数据，需要使用create_agent方法来创建agent的，否则无法获取到上下文信息，因为State等都是LangGraph中的概念，而create_agent的底层就是使用LangGraph来构建的，因此如果不使用create_agent创建Agent，而是直接调用model或者直接调用工具是拿不到对应信息的。

2.1 获取信息范围

LangChain给出了一张ToolRuntime可以访问信息的示意图：

主要包含以下几个部分：

Context（上下文）：运行时传入的固定配置，比如在创建Agent时传入的用户id、用户权限信息等内容。

LangGraph State（短期记忆）：在State内部保存Agent的短期记忆信息，和用户自己定义的短期记忆数据。

LangGraph Store（长期记忆）：在Store中的数据，是可以跨会话存储的，可以用来保存Agent长期记忆，在生产环境一般会将Store的数据持久化到数据库中。

Stream Writer（流写入器）：通过流写入器，可以在工具内向外部输出，当前工具的执行进度，尤其是处理大文件等耗时很长的工具。

Execution Info：可以获取LangGraph的thread_id等信息。

2.2 获取State数据

通过ToolRuntime获取State数据。

import dotenv
from langchain.agents import create_agent
from langchain.messages import HumanMessage
from langchain.tools import tool, ToolRuntime

dotenv.load_dotenv()


# 1.定义工具
@tool
def search_weather(runtime: ToolRuntime, city: str):
    """查询今日天气"""
    print(runtime.state["messages"])
    return f"今日{city}天气晴，东风4级，体感舒适"


# 2.创建Agent
agent = create_agent(model="deepseek-v4-flash",
                     system_prompt="你是一个专门为程序员工作的AI助手",
                     tools=[search_weather])

# 3.调用Agent
state = agent.invoke({
    "messages": [
        HumanMessage("今天杭州天气")
    ]
})

# 4.输出结果
print(state.get("messages")[-1].content)

执行结果如下：

2.3 更新State数据

可以在工具中返回Command对象对State中的数据进行修改，示例代码如下：

from datetime import datetime

import dotenv
from langchain.agents import create_agent, AgentState
from langchain.messages import HumanMessage, ToolMessage
from langchain.tools import tool, ToolRuntime
from langgraph.types import Command

dotenv.load_dotenv()


# 1.定义State
class CustomState(AgentState):
    last_search_weather_time: str


# 2.定义工具
@tool
def search_weather(runtime: ToolRuntime, city: str):
    """查询今日天气"""
    return Command(
        update={
            "messages": [
                ToolMessage(
                    content=f"今日{city}天气晴，东风4级，体感舒适",
                    tool_call_id=runtime.tool_call_id,
                )
            ],
            "last_search_weather_time": datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
        },
    )


# 3.创建Agent
agent = create_agent(model="deepseek-v4-flash",
                     system_prompt="你是一个专门为程序员工作的AI助手",
                     tools=[search_weather],
                     state_schema=CustomState, )

# 4.调用Agent
state = agent.invoke({
    "messages": [
        HumanMessage("今天杭州天气")
    ],
    "last_search_weather_time": None
})

# 5.输出结果
print(state.get("last_search_weather_time"))

执行结果：

2.4 获取context上下文

通过ToolRuntime可以访问应用的上下文信息，获取如当前用户信息等不可变信息，这里在调用agent时，还要传入thread_id，这样保证在同一个会话中，都可以获取到相同的上下文。代码示例如下：

import uuid
from dataclasses import dataclass

import dotenv
from langchain.agents import create_agent
from langchain.messages import HumanMessage
from langchain.tools import tool, ToolRuntime

dotenv.load_dotenv()


# 1.定义用户上下文对象
@dataclass
class UserContext:
    user_id: str


# 2.定义工具
@tool
def get_user_info(runtime: ToolRuntime[UserContext]):
    """获取当前用户信息"""
    # 忽略查询数据库具体操作...
    return f"用户id：{runtime.context.user_id}, 用户名：大志"


# 3.创建Agent
agent = create_agent(model="deepseek-v4-flash",
                     system_prompt="你是一个专门为程序员工作的AI助手",
                     context_schema=UserContext,
                     tools=[get_user_info])

# 4.调用Agent
state = agent.invoke(
    {"messages": [HumanMessage("获取当前用户信息")]},
    context=UserContext(user_id="123"),
    config={"configurable": {"thread_id": str(uuid.uuid4())}}, )

# 5.输出结果
print(state.get("messages")[-1].content)

执行结果：

2.5 获取长期记忆

还可以通过ToolRuntime获取长期记忆，在不同的会话中，长期记忆都可以被读取到，在生产环境长期记忆一般会存储到数据库中，这里使用基于内存存储的InMemoryStore进行演示。

代码示例如下，其中元组users是命名空间，id是存储字典的key，user_info是存储用户数据的value。

import dotenv
from langchain.agents import create_agent
from langchain.messages import HumanMessage
from langchain.tools import tool, ToolRuntime
from langgraph.store.memory import InMemoryStore
from pydantic import BaseModel, Field

dotenv.load_dotenv()


# 1.定义用户信息
class UserInfo(BaseModel):
    """用户信息"""
    id: str = Field(description="用户id")
    name: str = Field(description="用户名")
    sex: str = Field(description="性别")


# 2.定义工具信息
@tool
def get_user_info(runtime: ToolRuntime, user_id: str):
    """获取用户信息"""
    return runtime.store.get(("users",), user_id)


@tool(args_schema=UserInfo)
def add_user_info(runtime: ToolRuntime, id: str, name: str, sex: str):
    """添加用户信息"""
    return runtime.store.put(("users",), id, UserInfo(id=id, name=name, sex=sex))


store = InMemoryStore()
# 3.创建Agent
agent = create_agent(model="deepseek-v4-flash",
                     system_prompt="你是一个专门为程序员工作的AI助手",
                     tools=[get_user_info, add_user_info],
                     store=store)

# 4.调用Agent
state = agent.invoke({
    "messages": [
        HumanMessage("保存用户信息，用户id：1，用户名：张三，性别：男")
    ]
})

state = agent.invoke({
    "messages": [
        HumanMessage("获取用户id为1的用户信息")
    ]
})

# 5.输出结果
print(state.get("messages")[-1].content)

执行结果：

2.6 获取流写入器

在一些处理文件或者处理复杂任务的工具中，往往耗时非常长，所以在工具内部最好能实时的输出处理的进度，这样使用体验会好很多，可以通过ToolRuntime，获取对应的writer对象，在执行过程中，通过流式输出输出进度。

这里需要注意的是，首先调用agent要使用stream流式输出，并且设置输出模式为update和custom，只有设置了custom才能接收到，我们在工具中输出的自定义流式消息。

import dotenv
import time
from langchain.agents import create_agent
from langchain.messages import HumanMessage
from langchain.tools import tool, ToolRuntime

dotenv.load_dotenv()


# 1.定义工具
@tool
def generate_ppt(runtime: ToolRuntime, subject: str):
    """生成ppt"""
    writer = runtime.stream_writer
    writer(f"第1步：开始解析主题{subject}")
    time.sleep(3)
    writer(f"第2步：查找模板库")
    time.sleep(3)
    writer(f"第3步：布局排版")
    time.sleep(3)
    writer(f"第4步：最终检查")
    time.sleep(3)
    writer(f"第5步：输出文件")

    return f"{subject}.ppt"


# 2.创建Agent
agent = create_agent(model="deepseek-v4-flash",
                     system_prompt="你是一个专门为程序员工作的AI助手",
                     tools=[generate_ppt])

# 3.调用Agent
for chunk in agent.stream({
    "messages": [
        HumanMessage("生成一个主题为儿童节的ppt")
    ]
}, stream_mode=["updates", "custom"]):
    print(chunk)

执行结果：

2.7 获取执行信息

通过ToolRuntime还可以获得执行信息，如当前会话的线程id、节点执行失败后的重试次数等信息，代码示例如下：

import dotenv
from langchain.agents import create_agent
from langchain.messages import HumanMessage
from langchain.tools import tool, ToolRuntime

dotenv.load_dotenv()


# 1.定义工具
@tool
def search_weather(runtime: ToolRuntime, city: str):
    """查询今日天气"""
    info = runtime.execution_info

    print(f"线程id: {info.thread_id}")
    print(f"重试次数: {info.node_attempt}")
    return f"今日{city}天气晴，东风4级，体感舒适"


# 2.创建Agent
agent = create_agent(model="deepseek-v4-flash",
                     system_prompt="你是一个专门为程序员工作的AI助手",
                     tools=[search_weather])

# 3.调用Agent
state = agent.invoke({
    "messages": [
        HumanMessage("今天杭州天气")
    ]
}, config={"configurable": {"thread_id": "1234"}},)

执行结果：

三、执行工具

3.1 工具的返回值

工具的返回值可以包括以下几种：

• 字符串：返回的字符串会被LangChain包装成一个ToolMessage
• 对象：返回的对象会被序列化之后，再通过ToolMessage进行包装。
• Comand对象：当需要更新LangGraph中的State数据时，可以利用Command，并且需要自行组装ToolMessage对象。

工具返回字符串：

@tool("search_ip_location", description="通过用户提供的ip地址，获取对应地理位置")
def get_ip_location(ip: str) -> str:
    """通过ip地址获取位置信息

    Args:
        ip: 要查询定位的ip地址
    """
    return f"ip地址：{ip}，定位地址：浙江杭州"

返回对象：

# 2.返回对象的工具
@tool("get_user_info", description="根据用户id，获取用户信息")
def get_user_info(user_id: str) -> dict[str, any]:
    return {
        "id": user_id,
        "name": "李四",
        "sex": "女",
        "phone": "131xxxxxxxxx"
    }

返回Command对象

@tool("search_weather", description="通过传递的城市信息查询天气")
def search_weather(runtime: ToolRuntime, city: str):
    """查询今日天气"""
    return Command(
        update={
            "messages": [
                ToolMessage(
                    content=f"今日{city}天气晴，东风4级，体感舒适",
                    tool_call_id=runtime.tool_call_id,
                )
            ],
            "last_search_weather_time": datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
        },
    )

具体的用法在前面的案例中，已经介绍过了，这里就重复赘述了。

3.2 直接返回结果

一般情况下，无论是我们自己创建的Agent，还是利用LangChain提供的create_agent去创建Agent，都需要创建一个Agent循环，去循环调用工具直到完成任务，在工具调用完成之后，会把工具调用结果在交给LLM进行加工再返回。

比方说工具返回的内容是：

{
  "city": "杭州",
  "weather": "晴，气温30C°~35C°"
}

LLM会再进行加工再将内容返回：

根据天气信息，杭州今天晴天，30C°~35C°度，注意防暑。

但有时我们不想让LLM帮我们多加工一层，而是将工具调用结果直接返回给用户，那么就可以使用@tool(return_direct=True)让工具直接返回结果，代码示例如下：

import dotenv
from langchain.tools import tool
from langchain.agents import create_agent
from langchain.messages import HumanMessage

dotenv.load_dotenv()


# 1.定义工具
@tool("search_weather", description="根据城市名称查询天气信息", return_direct=True)
def search_weather(city: str):
    return {
        "city": city,
        "weather": "晴，气温30C°~35C°"
    }


# 2.创建Agent
agent = create_agent(model="deepseek-v4-flash",
                     system_prompt="你是一个天气查询助手",
                     tools=[search_weather])

# 3.调用Agent
state = agent.invoke({"messages": [HumanMessage("查询杭州天气")]})

# 4.输出结果
print(type(state.get("messages")[-1]))
print(state.get("messages")[-1].content)

执行结果如下，通过结果可以看出在执行完工具之后，就没有再去调用LLM，因此消息列表的最后一条消息是ToolMessage消息。

在使用直接返回工具内容时有一些注意点：

• 如果LLM返回多个工具调用，必须全部都设置成直接返回，才会直接返回工具结果。
• 如果先调用A工具，再调用LLM，然后再调用B工具，如果A工具设置了直接返回，那么后续的所有流程都不会再继续执行，B工具也就没有调用的机会了。

四、模型服务端内置工具

除了在Agent上配置的工具之外，现在一些LLM还提供了模型的内置工具，比如联网搜索工具、代码执行工具等等，下面以Qwen3.7-max来举例，使用它内部的联网搜索工具。

Qwen3.7-max提供的内置工具如下：

首先，在项目中安装对应依赖：

uv add langchain-qwq

使用qwen的内置工具仅需要传入额外的参数extra_body，并且开启联网搜索enable_search和网页抓取web_extractor，示例代码如下:

import dotenv
from langchain.agents import create_agent
from langchain.messages import HumanMessage
from langchain_qwq.chat_models import ChatQwen

dotenv.load_dotenv()

# 1.创建model
llm = ChatQwen(
    model="qwen3.7-max",
    extra_body={
        "enable_search": True,
        "web_extractor": True,
    }
)

# 2.创建Agent
agent = create_agent(model=llm)

# 3.调用Agent
state = agent.invoke({"messages": [HumanMessage("获取今日AI新闻")]})

# 4.输出结果
print(state.get("messages")[-1].content)

执行结果如下：